package com.mattgarner.jaddas.node.command.dml;

import java.io.ByteArrayInputStream;
import java.io.DataInputStream;

import com.mattgarner.jaddas.dataset.DataSet;
import com.mattgarner.jaddas.node.DataSetManager;
import com.mattgarner.jaddas.node.RemoteClient;
import com.mattgarner.jaddas.node.command.CommandParser;
import com.mattgarner.jaddas.node.data.LocalTableDataProvider;
import com.mattgarner.jaddas.node.data.TableHeader;
import com.mattgarner.jaddas.node.net.Message;
import com.mattgarner.jaddas.node.net.Protocol;
import com.mattgarner.jaddas.node.net.Worker;

public class CmdBulkLoader extends CommandParser {

	public CmdBulkLoader(Worker serverWorker, RemoteClient client) {
		super(serverWorker, client);
	}
	
	public final boolean processCommand(Message message) {
		Worker worker = getServerWorker();
		RemoteClient clientConfig = getRemoteClient();

		// decode binary packet containing multiple rows
		ByteArrayInputStream baisData = new ByteArrayInputStream(message.getMessageBytes());
		DataInputStream data = new DataInputStream(baisData);
		
		String dsName = null;
		short columnCount = 0;
		short selectedColumnCount = 0;
		String[] selectedColumnNames = null;
		short linesExpected = 0;
		int rowsImported = 0;
		
		try {
			
			// headers
			dsName = data.readUTF();
			
			DataSetManager dsManager = DataSetManager.getInstance();
			DataSet dataSet = dsManager.getDataSetByName(dsName);
			
			if(dataSet == null) {
				worker.sendMessage(Protocol.MSG_NAK, "DataSet '"+dsName+"' does not exist.");
				return true;
			}
			
			LocalTableDataProvider dsDataProvider = new LocalTableDataProvider(dataSet);
			TableHeader dataTableHeader = dataSet.getDataTableHeader();
			
			columnCount = data.readShort();
			selectedColumnCount = data.readShort();
			
			selectedColumnNames = new String[selectedColumnCount];
			for(int a=0; a<selectedColumnCount; a++) {
				selectedColumnNames[a] = data.readUTF();
			}
			linesExpected = data.readShort();
			
			// extract binary representation of each row
			for(short b=0; b<linesExpected; b++) {
				short rowNumber = data.readShort();
				if(rowNumber != b) {
					worker.sendMessage(Protocol.MSG_NAK, "Row data binary misalignment.");
					return true;
				}
				
				// decode separate column values
				if((selectedColumnCount>0) && (selectedColumnCount<columnCount)) {
					columnCount = selectedColumnCount;
				}
				String[] columnFieldData = new String[columnCount];
				for(short c=0; c<columnCount; c++) {
					columnFieldData[c] = data.readUTF();
				}
				
				String[] columns = new String[dataTableHeader.getColumnCount()];
				String[] values = new String[dataTableHeader.getColumnCount()];
				
				if((columnFieldData.length < dataTableHeader.getColumnCount()) || (selectedColumnCount > 0))  {
					// TODO: we have a subset of the fields
				} else if(columnFieldData.length > dataTableHeader.getColumnCount()) {
					worker.sendMessage(Protocol.MSG_NAK, "Too many columns specified for '"+dsName+"'.");
					return true;
				} else {
					// order data for insert (one to one)
					for(short d=0; d<dataTableHeader.getColumnCount(); d++) {
						columns[d] = dataTableHeader.getColumnName(d);
						values[d] = columnFieldData[d];
					}
				}
									
				dsDataProvider.updateIntoDataSet(columns, values);
				rowsImported++;
				
			}
			worker.sendMessage(Protocol.MSG_OK, "Imported "+rowsImported+" rows into '"+dsName+"'. (in "+calculateExecuteTime()+" secs.)");
		} catch (Exception e) {
			worker.sendMessage(Protocol.MSG_NAK, "Error: " + e.getMessage());
			
		}
		return true;

	}

}
